package cn.jly.bigdata.spark.sql_practice

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

/**
 * @author lanyangji
 * @date 2019/12/2 17:03
 */
object SparkSql_Practice {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .config(new SparkConf())
      .master("local[*]")
      .appName("SparkSql_Practice")
      .getOrCreate()

    import spark.implicits._

    println("------------ tbStock ------------------")
    /*
      BYSL00000893,ZHAO,2007-8-23
      BYSL00000897,ZHAO,2007-8-24
      BYSL00000898,ZHAO,2007-8-25
     */
    val tbStockRdd: RDD[String] = spark.sparkContext.textFile("input/good/tbStock.txt")
    val tbStockDS: Dataset[TbStock] = tbStockRdd
      .map(_.split(","))
      .map(attr => TbStock(attr(0), attr(1), attr(2)))
      .toDS

    // 查询tbStock全部记录
    tbStockDS.show()

    println("------------ tbStockDetail ------------------")
    /*
      BYSL00000893,0,FS527258160501,-1,268.0,-268.0
      BYSL00000893,1,FS527258169701,1,268.0,268.0
     */
    val tbStockDetailRdd: RDD[String] = spark.sparkContext.textFile("input/good/tbStockDetail.txt")
    val tbStockDetailDS: Dataset[TbStockDetail] = tbStockDetailRdd
      .map(_.split(","))
      .map {
        case Array(orderNumber, rowNum, itemId, number, price, amount) =>
          TbStockDetail(orderNumber, rowNum.trim.toInt, itemId,
            number.trim.toInt, price.trim.toDouble, amount.trim.toDouble)
      }.toDS

    // 打印 tbStockDetail 表全部记录
    tbStockDetailDS.show()

    println("------------ tbDate ------------------")
    /*
      2003-1-1,200301,2003,1,1,3,1,1,1,1
      2003-1-2,200301,2003,1,2,4,1,1,1,1
     */
    val tbDateRdd: RDD[String] = spark.sparkContext.textFile("input/good/tbDate.txt")
    val tbDateDS: Dataset[TbDate] = tbDateRdd
      .map(_.split(","))
      .map {
        case Array(dateId, years, theYear, month, day, weekday, week, quarter, period, halfMonth) =>
          TbDate(dateId, years.trim.toInt, theYear.trim.toInt,
            month.trim.toInt, day.trim.toInt, weekday.trim.toInt,
            week.trim.toInt, quarter.trim.toInt, period.trim.toInt,
            halfMonth.trim.toInt)
      }.toDS

    // 打印日期表所有记录
    tbDateDS.show()

    println("------------ 建立临时表 ------------------")
    tbStockDS.createOrReplaceTempView("tbStock")
    tbStockDetailDS.createOrReplaceTempView("tbStockDetail")
    tbDateDS.createOrReplaceTempView("tbDate")

    println("------------ 打印当前有哪些表 ------------------")
    spark.sql("show tables").show()

    println("------------ 统计所有订单中，每年的销售单数，销售总额 ------------------")

    spark.sql("select * from tbStock").show()
    spark.sql("select * from tbStockDetail").show()
    spark.sql("select * from tbDate").show()

    // 释放资源
    spark.close()
  }
}
